﻿using System;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Configuration;
using System.Globalization;
using System.Text;

namespace sBlog.Net.Memcache
{
    public class MemcachedClient
    {
        private static Dictionary<string, MemcachedClient> instances = new Dictionary<string, MemcachedClient>();
        private static LogAdapter logger = LogAdapter.GetLogger(typeof(MemcachedClient));

        public static void Setup(string name, string[] servers)
        {
            if (instances.ContainsKey(name))
                throw new ConfigurationErrorsException("Trying to configure MemcachedClient instance \"" + name + "\" twice.");
            instances[name] = new MemcachedClient(name, servers);
        }
        
        public static bool Exists(string name)
        {
            return instances.ContainsKey(name);
        }
        
        private static MemcachedClient defaultInstance = null;

        public static MemcachedClient GetInstance()
        {
            return defaultInstance ?? (defaultInstance = GetInstance("default"));
        }
  
        public static MemcachedClient GetInstance(string name)
        {
            MemcachedClient c;

            if (instances.TryGetValue(name, out c))
            {
                return c;
            }
            else
            {
                NameValueCollection config = ConfigurationManager.GetSection("beitmemcached") as NameValueCollection;
                if (config != null && !String.IsNullOrEmpty(config.Get(name)))
                {
                    Setup(name, config.Get(name).Split(new char[] { ',' }));
                    return GetInstance(name);
                }
                throw new ConfigurationErrorsException("Unable to find MemcachedClient instance \"" + name + "\".");
            }
        }
       
        public readonly string Name;
        private readonly ServerPool serverPool;        

        public string KeyPrefix { get { return keyPrefix; } set { keyPrefix = value; } }
        private string keyPrefix = "";        
        public int SendReceiveTimeout { get { return serverPool.SendReceiveTimeout; } set { serverPool.SendReceiveTimeout = value; } }
        
        public uint MinPoolSize
        {
            get { return serverPool.MinPoolSize; }
            set
            {
                if (value > MaxPoolSize) { throw new ConfigurationErrorsException("MinPoolSize (" + value + ") may not be larger than the MaxPoolSize (" + MaxPoolSize + ")."); }
                serverPool.MinPoolSize = value;
            }
        }
    
        public uint MaxPoolSize
        {
            get { return serverPool.MaxPoolSize; }
            set
            {
                if (value < MinPoolSize) { throw new ConfigurationErrorsException("MaxPoolSize (" + value + ") may not be smaller than the MinPoolSize (" + MinPoolSize + ")."); }
                serverPool.MaxPoolSize = value;
            }
        }
    
        public TimeSpan SocketRecycleAge { get { return serverPool.SocketRecycleAge; } set { serverPool.SocketRecycleAge = value; } }

        private uint compressionThreshold = 1024 * 128; //128kb
        
        public uint CompressionThreshold { get { return compressionThreshold; } set { compressionThreshold = value; } }
        
        private MemcachedClient(string name, string[] hosts)
        {
            if (String.IsNullOrEmpty(name))            
                throw new ConfigurationErrorsException("Name of MemcachedClient instance cannot be empty.");
            
            if (hosts == null || hosts.Length == 0)            
                throw new ConfigurationErrorsException("Cannot configure MemcachedClient with empty list of hosts.");            

            Name = name;
            serverPool = new ServerPool(hosts);
        }
     
        private uint hash(string key)
        {
            checkKey(key);
            return BitConverter.ToUInt32(new ModifiedFNV1_32().ComputeHash(Encoding.UTF8.GetBytes(key)), 0);
        }
        
        private uint hash(uint hashvalue)
        {
            return BitConverter.ToUInt32(new ModifiedFNV1_32().ComputeHash(BitConverter.GetBytes(hashvalue)), 0);
        }
     
        private uint[] hash(string[] keys)
        {
            uint[] result = new uint[keys.Length];
            for (int i = 0; i < keys.Length; i++)
            {
                result[i] = hash(keys[i]);
            }
            return result;
        }
     
        private uint[] hash(uint[] hashvalues)
        {
            uint[] result = new uint[hashvalues.Length];
            for (int i = 0; i < hashvalues.Length; i++)
            {
                result[i] = hash(hashvalues[i]);
            }
            return result;
        }
    
        private void checkKey(string key)
        {
            if (key == null)            
                throw new ArgumentNullException("Key may not be null.");
            
            if (key.Length == 0)            
                throw new ArgumentException("Key may not be empty.");
            
            if (key.Length > 250)            
                throw new ArgumentException("Key may not be longer than 250 characters.");
            
            if (key.Contains(" ") || key.Contains("\n") || key.Contains("\r") || key.Contains("\t") || key.Contains("\f") || key.Contains("\v"))            
                throw new ArgumentException("Key may not contain whitespace or control characters.");            
        }

        private static DateTime epoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);

        private static int getUnixTime(DateTime datetime)
        {
            return (int)(datetime.ToUniversalTime() - epoch).TotalSeconds;
        }

        public bool Set(string key, object value) { return store("set", key, true, value, hash(key), 0); }
        public bool Set(string key, object value, uint hash) { return store("set", key, false, value, this.hash(hash), 0); }
        public bool Set(string key, object value, TimeSpan expiry) { return store("set", key, true, value, hash(key), (int)expiry.TotalSeconds); }
        public bool Set(string key, object value, uint hash, TimeSpan expiry) { return store("set", key, false, value, this.hash(hash), (int)expiry.TotalSeconds); }
        public bool Set(string key, object value, DateTime expiry) { return store("set", key, true, value, hash(key), getUnixTime(expiry)); }
        public bool Set(string key, object value, uint hash, DateTime expiry) { return store("set", key, false, value, this.hash(hash), getUnixTime(expiry)); }
       
        public bool Add(string key, object value) { return store("add", key, true, value, hash(key), 0); }
        public bool Add(string key, object value, uint hash) { return store("add", key, false, value, this.hash(hash), 0); }
        public bool Add(string key, object value, TimeSpan expiry) { return store("add", key, true, value, hash(key), (int)expiry.TotalSeconds); }
        public bool Add(string key, object value, uint hash, TimeSpan expiry) { return store("add", key, false, value, this.hash(hash), (int)expiry.TotalSeconds); }
        public bool Add(string key, object value, DateTime expiry) { return store("add", key, true, value, hash(key), getUnixTime(expiry)); }
        public bool Add(string key, object value, uint hash, DateTime expiry) { return store("add", key, false, value, this.hash(hash), getUnixTime(expiry)); }
   
        public bool Replace(string key, object value) { return store("replace", key, true, value, hash(key), 0); }
        public bool Replace(string key, object value, uint hash) { return store("replace", key, false, value, this.hash(hash), 0); }
        public bool Replace(string key, object value, TimeSpan expiry) { return store("replace", key, true, value, hash(key), (int)expiry.TotalSeconds); }
        public bool Replace(string key, object value, uint hash, TimeSpan expiry) { return store("replace", key, false, value, this.hash(hash), (int)expiry.TotalSeconds); }
        public bool Replace(string key, object value, DateTime expiry) { return store("replace", key, true, value, hash(key), getUnixTime(expiry)); }
        public bool Replace(string key, object value, uint hash, DateTime expiry) { return store("replace", key, false, value, this.hash(hash), getUnixTime(expiry)); }
    
        public bool Append(string key, object value) { return store("append", key, true, value, hash(key)); }
        public bool Append(string key, object value, uint hash) { return store("append", key, false, value, this.hash(hash)); }
    
        public bool Prepend(string key, object value) { return store("prepend", key, true, value, hash(key)); }
        public bool Prepend(string key, object value, uint hash) { return store("prepend", key, false, value, this.hash(hash)); }

        public enum CasResult
        {
            Stored = 0,
            NotStored = 1,
            Exists = 2,
            NotFound = 3
        }

        public CasResult CheckAndSet(string key, object value, ulong unique) { return store(key, true, value, hash(key), 0, unique); }
        public CasResult CheckAndSet(string key, object value, uint hash, ulong unique) { return store(key, false, value, this.hash(hash), 0, unique); }
        public CasResult CheckAndSet(string key, object value, TimeSpan expiry, ulong unique) { return store(key, true, value, hash(key), (int)expiry.TotalSeconds, unique); }
        public CasResult CheckAndSet(string key, object value, uint hash, TimeSpan expiry, ulong unique) { return store(key, false, value, this.hash(hash), (int)expiry.TotalSeconds, unique); }
        public CasResult CheckAndSet(string key, object value, DateTime expiry, ulong unique) { return store(key, true, value, hash(key), getUnixTime(expiry), unique); }
        public CasResult CheckAndSet(string key, object value, uint hash, DateTime expiry, ulong unique) { return store(key, false, value, this.hash(hash), getUnixTime(expiry), unique); }

        private bool store(string command, string key, bool keyIsChecked, object value, uint hash, int expiry)
        {
            return store(command, key, keyIsChecked, value, hash, expiry, 0).StartsWith("STORED");
        }

        private bool store(string command, string key, bool keyIsChecked, object value, uint hash)
        {
            return store(command, key, keyIsChecked, value, hash, 0, 0).StartsWith("STORED");
        }

        private CasResult store(string key, bool keyIsChecked, object value, uint hash, int expiry, ulong unique)
        {
            string result = store("cas", key, keyIsChecked, value, hash, expiry, unique);
            if (result.StartsWith("STORED"))
            {
                return CasResult.Stored;
            }
            else if (result.StartsWith("EXISTS"))
            {
                return CasResult.Exists;
            }
            else if (result.StartsWith("NOT_FOUND"))
            {
                return CasResult.NotFound;
            }
            return CasResult.NotStored;
        }

        private string store(string command, string key, bool keyIsChecked, object value, uint hash, int expiry, ulong unique)
        {
            if (!keyIsChecked)            
                checkKey(key);
            
            return serverPool.Execute<string>(hash, "", delegate(PooledSocket socket)
            {
                SerializedType type;
                byte[] bytes;

                try
                {
                    bytes = Serializer.Serialize(value, out type, CompressionThreshold);
                }
                catch (Exception e)
                {
                    logger.Error("Error serializing object for key '" + key + "'.", e);
                    return "";
                }

                string commandline = "";
                switch (command)
                {
                    case "set":
                    case "add":
                    case "replace":
                        commandline = command + " " + keyPrefix + key + " " + (ushort)type + " " + expiry + " " + bytes.Length + "\r\n";
                        break;
                    case "append":
                    case "prepend":
                        commandline = command + " " + keyPrefix + key + " 0 0 " + bytes.Length + "\r\n";
                        break;
                    case "cas":
                        commandline = command + " " + keyPrefix + key + " " + (ushort)type + " " + expiry + " " + bytes.Length + " " + unique + "\r\n";
                        break;
                }

                socket.Write(commandline);
                socket.Write(bytes);
                socket.Write("\r\n");
                return socket.ReadResponse();
            });
        }
  
        public object Get(string key) { ulong i; return get("get", key, true, hash(key), out i); }
        public object Get(string key, uint hash) { ulong i; return get("get", key, false, this.hash(hash), out i); }
        public object Gets(string key, out ulong unique) { return get("gets", key, true, hash(key), out unique); }
        public object Gets(string key, uint hash, out ulong unique) { return get("gets", key, false, this.hash(hash), out unique); }

        private object get(string command, string key, bool keyIsChecked, uint hash, out ulong unique)
        {
            if (!keyIsChecked)            
                checkKey(key);
          
            ulong __unique = 0;
            object value = serverPool.Execute<object>(hash, null, delegate(PooledSocket socket)
            {
                socket.Write(command + " " + keyPrefix + key + "\r\n");
                object _value;
                ulong _unique;
                if (readValue(socket, out _value, out key, out _unique))
                {
                    socket.ReadLine(); 
                }
                __unique = _unique;
                return _value;
            });
            unique = __unique;
            return value;
        }
        
        public object[] Get(string[] keys) { ulong[] uniques; return get("get", keys, true, hash(keys), out uniques); }
        public object[] Get(string[] keys, uint[] hashes) { ulong[] uniques; return get("get", keys, false, hash(hashes), out uniques); }        
        public object[] Gets(string[] keys, out ulong[] uniques) { return get("gets", keys, true, hash(keys), out uniques); }
        public object[] Gets(string[] keys, uint[] hashes, out ulong[] uniques) { return get("gets", keys, false, hash(hashes), out uniques); }
        private object[] get(string command, string[] keys, bool keysAreChecked, uint[] hashes, out ulong[] uniques)
        {            
            if (keys == null || hashes == null)            
                throw new ArgumentException("Keys and hashes arrays must not be null.");
            
            if (keys.Length != hashes.Length)            
                throw new ArgumentException("Keys and hashes arrays must be of the same length.");
            
            uniques = new ulong[keys.Length];
            
            if (keys.Length == 1)            
                return new object[] { get(command, keys[0], keysAreChecked, hashes[0], out uniques[0]) };

            if (!keysAreChecked)
            {
                for (int i = 0; i < keys.Length; i++)
                {
                    checkKey(keys[i]);
                }
            }
            
            Dictionary<SocketPool, Dictionary<string, List<int>>> dict = new Dictionary<SocketPool, Dictionary<string, List<int>>>();
            for (int i = 0; i < keys.Length; i++)
            {
                Dictionary<string, List<int>> getsForServer;
                SocketPool pool = serverPool.GetSocketPool(hashes[i]);
                if (!dict.TryGetValue(pool, out getsForServer))                
                    dict[pool] = getsForServer = new Dictionary<string, List<int>>();
              
                List<int> positions;

                if (!getsForServer.TryGetValue(keys[i], out positions))                
                    getsForServer[keys[i]] = positions = new List<int>();
                
                positions.Add(i);
            }

            object[] returnValues = new object[keys.Length];
            ulong[] _uniques = new ulong[keys.Length];
            foreach (KeyValuePair<SocketPool, Dictionary<string, List<int>>> kv in dict)
            {
                serverPool.Execute(kv.Key, delegate(PooledSocket socket)
                {
                    StringBuilder getRequest = new StringBuilder(command);
                    foreach (KeyValuePair<string, List<int>> key in kv.Value)
                    {
                        getRequest.Append(" ");
                        getRequest.Append(keyPrefix);
                        getRequest.Append(key.Key);
                    }

                    getRequest.Append("\r\n");                   
                    socket.Write(getRequest.ToString());

                    object gottenObject;
                    string gottenKey;
                    ulong unique;
                    while (readValue(socket, out gottenObject, out gottenKey, out unique))
                    {
                        foreach (int position in kv.Value[gottenKey])
                        {
                            returnValues[position] = gottenObject;
                            _uniques[position] = unique;
                        }
                    }
                });
            }
            uniques = _uniques;
            return returnValues;
        }

        private bool readValue(PooledSocket socket, out object value, out string key, out ulong unique)
        {
            string response = socket.ReadResponse();
            string[] parts = response.Split(' '); 
            if (parts[0] == "VALUE")
            {
                key = parts[1];
                SerializedType type = (SerializedType)Enum.Parse(typeof(SerializedType), parts[2]);
                byte[] bytes = new byte[Convert.ToUInt32(parts[3], CultureInfo.InvariantCulture)];
                if (parts.Length > 4)
                {
                    unique = Convert.ToUInt64(parts[4]);
                }
                else
                {
                    unique = 0;
                }
                socket.Read(bytes);
                socket.SkipUntilEndOfLine(); 
                try
                {
                    value = Serializer.DeSerialize(bytes, type);
                }
                catch (Exception e)
                {                    
                    value = null;
                    logger.Error("Error deserializing object for key '" + key + "' of type " + type + ".", e);
                }
                return true;
            }
            else
            {
                key = null;
                value = null;
                unique = 0;
                return false;
            }
        }
     
        public bool Delete(string key) { return delete(key, true, hash(key), 0); }
        public bool Delete(string key, uint hash) { return delete(key, false, this.hash(hash), 0); }
        public bool Delete(string key, TimeSpan delay) { return delete(key, true, hash(key), (int)delay.TotalSeconds); }
        public bool Delete(string key, uint hash, TimeSpan delay) { return delete(key, false, this.hash(hash), (int)delay.TotalSeconds); }
        public bool Delete(string key, DateTime delay) { return delete(key, true, hash(key), getUnixTime(delay)); }
        public bool Delete(string key, uint hash, DateTime delay) { return delete(key, false, this.hash(hash), getUnixTime(delay)); }

        private bool delete(string key, bool keyIsChecked, uint hash, int time)
        {
            if (!keyIsChecked)            
                checkKey(key);            

            return serverPool.Execute<bool>(hash, false, delegate(PooledSocket socket)
            {
                string commandline;
                if (time == 0)
                {
                    commandline = "delete " + keyPrefix + key + "\r\n";
                }
                else
                {
                    commandline = "delete " + keyPrefix + key + " " + time + "\r\n";
                }
                socket.Write(commandline);
                return socket.ReadResponse().StartsWith("DELETED");
            });
        }

        public bool SetCounter(string key, ulong value) { return Set(key, value.ToString(CultureInfo.InvariantCulture)); }
        public bool SetCounter(string key, ulong value, uint hash) { return Set(key, value.ToString(CultureInfo.InvariantCulture), this.hash(hash)); }
        public bool SetCounter(string key, ulong value, TimeSpan expiry) { return Set(key, value.ToString(CultureInfo.InvariantCulture), expiry); }
        public bool SetCounter(string key, ulong value, uint hash, TimeSpan expiry) { return Set(key, value.ToString(CultureInfo.InvariantCulture), this.hash(hash), expiry); }
        public bool SetCounter(string key, ulong value, DateTime expiry) { return Set(key, value.ToString(CultureInfo.InvariantCulture), expiry); }
        public bool SetCounter(string key, ulong value, uint hash, DateTime expiry) { return Set(key, value.ToString(CultureInfo.InvariantCulture), this.hash(hash), expiry); }

        public ulong? GetCounter(string key) { return getCounter(key, true, hash(key)); }
        public ulong? GetCounter(string key, uint hash) { return getCounter(key, false, this.hash(hash)); }

        private ulong? getCounter(string key, bool keyIsChecked, uint hash)
        {
            ulong parsedLong, unique;
            return ulong.TryParse(get("get", key, keyIsChecked, hash, out unique) as string, out parsedLong) ? (ulong?)parsedLong : null;
        }

        public ulong?[] GetCounter(string[] keys) { return getCounter(keys, true, hash(keys)); }
        public ulong?[] GetCounter(string[] keys, uint[] hashes) { return getCounter(keys, false, hash(hashes)); }

        private ulong?[] getCounter(string[] keys, bool keysAreChecked, uint[] hashes)
        {
            ulong?[] results = new ulong?[keys.Length];
            ulong[] uniques;
            object[] values = get("get", keys, keysAreChecked, hashes, out uniques);
            for (int i = 0; i < values.Length; i++)
            {
                ulong parsedLong;
                results[i] = ulong.TryParse(values[i] as string, out parsedLong) ? (ulong?)parsedLong : null;
            }
            return results;
        }
   
        public ulong? Increment(string key, ulong value) { return incrementDecrement("incr", key, true, value, hash(key)); }
        public ulong? Increment(string key, ulong value, uint hash) { return incrementDecrement("incr", key, false, value, this.hash(hash)); }
      
        public ulong? Decrement(string key, ulong value) { return incrementDecrement("decr", key, true, value, hash(key)); }
        public ulong? Decrement(string key, ulong value, uint hash) { return incrementDecrement("decr", key, false, value, this.hash(hash)); }

        private ulong? incrementDecrement(string cmd, string key, bool keyIsChecked, ulong value, uint hash)
        {
            if (!keyIsChecked)
            {
                checkKey(key);
            }
            return serverPool.Execute<ulong?>(hash, null, delegate(PooledSocket socket)
            {
                string command = cmd + " " + keyPrefix + key + " " + value + "\r\n";
                socket.Write(command);
                string response = socket.ReadResponse();
                if (response.StartsWith("NOT_FOUND"))
                {
                    return null;
                }
                else
                {
                    return Convert.ToUInt64(response.TrimEnd('\0', '\r', '\n'));
                }
            });
        }
    
        public bool FlushAll() { return FlushAll(TimeSpan.Zero, false); }
        public bool FlushAll(TimeSpan delay) { return FlushAll(delay, false); }
        public bool FlushAll(TimeSpan delay, bool staggered)
        {
            bool noerrors = true;
            uint count = 0;
            foreach (SocketPool pool in serverPool.HostList)
            {
                serverPool.Execute(pool, delegate(PooledSocket socket)
                {
                    uint delaySeconds = (staggered ? (uint)delay.TotalSeconds * count : (uint)delay.TotalSeconds);
                    socket.Write("flush_all " + (delaySeconds == 0 ? "" : delaySeconds.ToString()) + "\r\n");

                    if (!socket.ReadResponse().StartsWith("OK"))                    
                        noerrors = false;
                    
                    count++;
                });
            }
            return noerrors;
        }
      
        public Dictionary<string, Dictionary<string, string>> Stats()
        {
            Dictionary<string, Dictionary<string, string>> results = new Dictionary<string, Dictionary<string, string>>();
            foreach (SocketPool pool in serverPool.HostList)
            {
                results.Add(pool.Host, stats(pool));
            }
            return results;
        }
   
        public Dictionary<string, string> Stats(string key) { return Stats(hash(key)); }
        public Dictionary<string, string> Stats(uint hash) { return stats(serverPool.GetSocketPool(this.hash(hash))); }
        public Dictionary<string, string> StatsByHost(string host) { return stats(serverPool.GetSocketPool(host)); }
        private Dictionary<string, string> stats(SocketPool pool)
        {
            if (pool == null)
            {
                return null;
            }
            Dictionary<string, string> result = new Dictionary<string, string>();
            serverPool.Execute(pool, delegate(PooledSocket socket)
            {
                socket.Write("stats\r\n");
                string line;
                while (!(line = socket.ReadResponse().TrimEnd('\0', '\r', '\n')).StartsWith("END"))
                {
                    string[] s = line.Split(' ');
                    result.Add(s[1], s[2]);
                }
            });
            return result;
        }
   
        public Dictionary<string, Dictionary<string, string>> Status()
        {
            Dictionary<string, Dictionary<string, string>> results = new Dictionary<string, Dictionary<string, string>>();
            foreach (SocketPool pool in serverPool.HostList)
            {
                Dictionary<string, string> result = new Dictionary<string, string>();
                if (serverPool.Execute<bool>(pool, false, delegate { return true; }))
                {
                    result.Add("Status", "Ok");
                }
                else
                {
                    result.Add("Status", "Dead, next retry at: " + pool.DeadEndPointRetryTime);
                }
                result.Add("Sockets in pool", pool.Poolsize.ToString());
                result.Add("Acquired sockets", pool.Acquired.ToString());
                result.Add("Sockets reused", pool.ReusedSockets.ToString());
                result.Add("New sockets created", pool.NewSockets.ToString());
                result.Add("New sockets failed", pool.FailedNewSockets.ToString());
                result.Add("Sockets died in pool", pool.DeadSocketsInPool.ToString());
                result.Add("Sockets died on return", pool.DeadSocketsOnReturn.ToString());
                result.Add("Dirty sockets on return", pool.DirtySocketsOnReturn.ToString());

                results.Add(pool.Host, result);
            }
            return results;
        }        
    }
}
